#  网站统计大数据
#  2017-12-27 20:00启动
#  参与人：壮志凌云

from __future__ import print_function
import re
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum,row_number,desc,count,countDistinct,from_unixtime,max,isnull,instr,col,regexp_replace,when
from pyspark.sql import Window
import json
from urllib import request


#  通过IP地址解析归属地和运营商
#  淘宝IP地址库不全，第一个IP就解析不出来
def checkip(list):
    dic = list.asDict()
    ip = dic["ip"]
    print(ip)
    URL = 'http://ip.taobao.com/service/getIpInfo.php'
    try:
        r = requests.get(URL, params=ip, timeout=3)
    except requests.RequestException as e:
        print(e)
    else:
        json_data = r.json()
        if json_data[u'code'] == 0:
            print('所在国家： ' + json_data[u'data'][u'country'])
            print('所在地区： ' + json_data[u'data'][u'area'])
            print('所在省份： ' + json_data[u'data'][u'region'])
            print('所在城市： ' + json_data[u'data'][u'city'])
            print('所属运营商：' + json_data[u'data'][u'isp'])
            dic["country"] = json_data[u'data'][u'country']
            dic["region"] = json_data[u'data'][u'region']
            dic["city"] = json_data[u'data'][u'city']
            dic["isp"] = json_data[u'data'][u'isp']
        else:
            print('查询失败,请稍后再试！')

#  使用IP138地址库解析
#  python2到python3变化 urllib2变为urllib.request re.findall时，要把r'变为b'
#  解析编解码，从网页获取的编码，经过调试百度后确认为使用.decode("gbk")解码。是python的asc11格式
#  巧用网址返回数据，使用re.split做分割，然后分别赋值做国家、省份、城市、运营商。避免复杂的算法
def ipcheck(list):
    dic = list.asDict()
    url = "http://www.ip138.com/ips138.asp?ip=%s&action=2" % dic["ip"]
#    url = "http://www.ip138.com/ips138.asp?ip=%s&action=2" % ipaddr
    u = request.urlopen(url)
    s = u.read()
    #Get IP Address
    ip = re.findall(b'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',s)
    print( "\n****** Below Result From IP138 Database *****")
    print("IP Address:",ip[0])
    #Get IP Address Location
    result = re.findall(b'(<li>.*?</li>)',s)
    print(re.split('：|:| ',result[0][4:-5].decode('gbk','ignore')))
    print(re.split('：|:| ',result[1][4:-5].decode('gbk','ignore')))
    print(re.split('：|:| ',result[2][4:-5].decode('gbk','ignore')))
#    dic["country"] = re.split('：|:| ',result[2][4:-5].decode('gbk','ignore'))[1]
    dic["city"] = re.split('：|:| ',result[0][4:-5].decode('gbk','ignore'))[1]
    dic["isp"] = re.split('：|:| ',result[0][4:-5].decode('gbk','ignore'))[3]
    print("*"*45)
    print("\n")
    return dic


if __name__ == "__main__":
    spark = SparkSession \
        .builder \
        .appName("网站统计-访客分析") \
        .getOrCreate()

#    ip = "117.136.72.89"
#    ipcheck(ip)

    #  采用python库函数，自行解析json，数据读取正常
    jsonData = spark.sparkContext.textFile("main_pv.json")
    WebData = jsonData.map(json.loads).toDF()
    WebData.printSchema()

    #  为数据添加日期、小时
    WebDataWithDH = WebData.withColumn('day', from_unixtime("vt", format='yyyy-MM-dd')).withColumn('hour',from_unixtime("vt",format='HH'))

    #  选取合适的指标，根据IP地址查找归属地和运营商信息
    #  添加网址来源信息，尝试新算法case when,在一列中添加网址来源信息,实验成功，为自己点个赞
    WebDataVisitor = WebDataWithDH.select(
        WebDataWithDH["day"],
        WebDataWithDH["hour"],
        WebDataWithDH["u"],
        WebDataWithDH["ip"],
        WebDataWithDH["b"],#浏览器
        WebDataWithDH["sc"],#屏幕分辨率
        WebDataWithDH["c"],#屏幕颜色
        WebDataWithDH["j"],#是否支持java
        WebDataWithDH["l"],#语言环境
        WebDataWithDH["k"],#是否支持Cookie
        WebDataWithDH["mi"], #设备类型
        when(
            WebDataWithDH["ref"].contains('baidu') |
            WebDataWithDH["ref"].contains('google') |
            WebDataWithDH["ref"].contains('bing.com') |
            WebDataWithDH["ref"].contains('so.com') |
            WebDataWithDH["ref"].contains('yahoo.com') |
            WebDataWithDH["ref"].contains('sougou.com'),"搜索引擎"
        ).otherwise(when(WebDataWithDH["ref"] == '',"直接访问").otherwise("外部链接")).alias("refClass")
    ).sample(False, 0.0001, 42).rdd.map(ipcheck).toDF() \
#    ).sample(False, 0.0001, 42).rdd.map(ipcheck).toDF() \  #测试用sample
#    .withColumn(
#        "refClassS",
#        WebDataWithDH["ref"].contains('baidu') |
#        WebDataWithDH["ref"].contains('google') |
#        WebDataWithDH["ref"].contains('bing.com') |
#        WebDataWithDH["ref"].contains('so.com') |
#        WebDataWithDH["ref"].contains('yahoo.com') |
#        WebDataWithDH["ref"].contains('sougou.com')
#    ).withColumn(
#        "refClassD",
#        WebDataWithDH["ref"] == ''
#    )

#    WebDataVisitor.printSchema()
#    WebDataVisitor.show()
#    print(WebDataVisitor.count())

    #  网页获取参数
    #  考虑到参数需要更新，list参数换成字典参数
    AnalysisCondition = {'day':"day = '2017-12-26'", 'hour':'', "u":'', "ip": '', "ref":'', "refClass" : '', "mi" : '', "url" : ''}
    #  来源类型和来源网站选择参数
    #  system取值为 浏览器、屏幕分辨率、屏幕颜色。。。。等等指标，作为groupby的参数
    visitorAnalysis = {"area":"city","system":"b"}

    #  DataCube Create
    #  有问题，在创建cube前，show（） count（）都是可以的，创建cube后，报错。猜测是使用的列太多了
    #  尝试删减列数，看是否可以cube。删减一部分列后，发现确实可以cube，并且cube后数据正常。
    #  是列多导致的还是本身cube承载数据量就不能大，成千上万的分析需求岂不是要崩溃
    #  使用分析什么数据，就用什么创建cube，剔除无关列
    WebDataVisitorCube = WebDataVisitor.cube(
        WebDataVisitor["day"],
#        WebDataVisitor["hour"],
#        WebDataVisitor["u"],
#        WebDataVisitor["ip"],
#        WebDataVisitor["b"],#浏览器
#        WebDataVisitor["sc"],#屏幕分辨率
#        WebDataVisitor["c"],#屏幕颜色
#        WebDataVisitor["j"],#是否支持java
#        WebDataVisitor["l"],#语言环境
#        WebDataVisitor["k"],#是否支持Cookie
#        WebDataVisitor["mi"], #设备类型
        WebDataVisitor[visitorAnalysis["area"]],
        WebDataVisitor[visitorAnalysis["system"]],
        WebDataVisitor["refClass"]
    ).count()

    WebDataVisitorCube.show()
    print(WebDataVisitorCube.count())

    ###################################################
    #  1.地域分布
    ###################################################
    WebDataVisitorPage = WebDataVisitorCube.filter(AnalysisCondition['day']) \
    .groupBy(visitorAnalysis["area"]) \
    .agg(
    max("count").alias("浏览量(PV)"),
#    countDistinct("u").alias("访客数(UV)"),
#    countDistinct("ip").alias("IP数")
    ).orderBy(desc("浏览量(PV)"))

    WebDataVisitorPage.show()

    ###################################################
    #  2.系统环境
    ###################################################
    WebDataVisitorSystem = WebDataVisitorCube.filter(AnalysisCondition['day']) \
    .groupBy(visitorAnalysis["system"]) \
    .agg(
    max("count").alias("浏览量(PV)"),
#    countDistinct("u").alias("访客数(UV)"),
#    countDistinct("ip").alias("IP数")
    ).orderBy(desc("浏览量(PV)"))

    WebDataVisitorSystem.show()

    ###################################################
    #  3.新老访客(无新老访客数据标识，暂时搁置)
    ###################################################

    ###################################################
    #  4.访客属性(无访客属性数据标识，暂时搁置)
    ###################################################

    ###################################################
    #  5.无忠诚度(无忠诚度数据标识，暂时搁置)
    ###################################################

    spark.stop()
